package com.newegg.hardware.benchmark.kafka.conf;

import java.util.Map;

import org.apache.kafka.common.errors.SerializationException;
import org.apache.kafka.common.serialization.Deserializer;
import com.alibaba.fastjson.JSONObject;

public class EntitySerializer implements Deserializer<Map<String, Object>> {
	private String encoding = "UTF8";
	
	@SuppressWarnings("unchecked")
	@Override
	public Map<String, Object> deserialize(String topic, byte[] data) {
		if (data == null) {
        	return null;
        } else {
        	String json = null;
        	try {
        		json = new String(data, encoding);
            	return JSONObject.parseObject(json, Map.class);
            } catch (Exception e) {
            	System.out.println(json);
                throw new SerializationException("Error when deserializing byte[] to string due to unsupported encoding " + encoding);
            }
        }
	}
}
